package handlers

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/gin-gonic/gin"
	"gorm.io/gorm"

	"task_queue/internal/models"
	"task_queue/internal/plugin/mq"
	"task_queue/internal/response"
)

// MessagePushRequest 定义消息推送请求结构
type MessagePushRequest struct {
	Topic   string    `json:"topic" binding:"required"`   // 主题名称
	MQType  mq.MQType `json:"mq_type" binding:"required"` // 消息队列类型
	Message string    `json:"message" binding:"required"` // 消息内容
}

// MessageConsumeRequest 定义消息消费请求结构
type MessageConsumeRequest struct {
	Topic  string    `json:"topic" binding:"required"`   // 主题名称
	MQType mq.MQType `json:"mq_type" binding:"required"` // 消息队列类型
	Count  int       `json:"count" binding:"required"`   // 消费消息数量
}

// MessageHandler 处理消息相关请求
type MessageHandler struct {
	db        *gorm.DB
	mqPlugins map[mq.MQType]mq.MQPlugin // 添加消息队列插件映射
}

// NewMessageHandler 创建消息处理器
func NewMessageHandler(db *gorm.DB, mqPlugins map[mq.MQType]mq.MQPlugin) *MessageHandler {
	return &MessageHandler{
		db:        db,
		mqPlugins: mqPlugins,
	}
}

// PushMessage 推送消息到指定队列
func (h *MessageHandler) PushMessage(c *gin.Context) {
	var req MessagePushRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		response.Fail(c, http.StatusBadRequest, "无效的请求参数: "+err.Error())
		return
	}

	// 从全局插件映射中获取对应类型的插件
	mqPlugin, ok := h.mqPlugins[req.MQType]
	if !ok {
		response.Fail(c, http.StatusBadRequest, "不支持的消息队列类型: "+string(req.MQType))
		return
	}

	// 推送消息
	ctx := c.Request.Context()
	if err := mqPlugin.Publish(ctx, req.Topic, req.Message); err != nil {
		response.Fail(c, http.StatusInternalServerError, "推送消息失败: "+err.Error())
		return
	}

	response.Success(c, gin.H{
		"message": "消息推送成功",
		"topic":   req.Topic,
		"mq_type": req.MQType,
	})
}

// GetAllMessages 获取所有消息
func (h *MessageHandler) GetAllMessages(c *gin.Context) {
	// 获取分页参数
	page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
	pageSize, _ := strconv.Atoi(c.DefaultQuery("pageSize", "10"))
	topic := c.Query("topic")
	mqType := c.Query("mq_type")

	if page < 1 {
		page = 1
	}
	if pageSize < 1 {
		pageSize = 10
	}
	offset := (page - 1) * pageSize

	// 查询任务表中的消息
	var tasks []models.Task
	query := h.db.Model(&models.Task{})

	// 添加过滤条件
	if topic != "" {
		query = query.Where("topic_name = ?", topic)
	}
	// 添加消息队列类型过滤
	if mqType != "" {
		query = query.Where("mq_type = ?", mqType)
	}

	// 查询分页数据
	result := query.Order("created_at DESC").Offset(offset).Limit(pageSize).Find(&tasks)
	if result.Error != nil {
		response.Fail(c, http.StatusInternalServerError, "查询消息列表失败: "+result.Error.Error())
		return
	}

	// 查询总数
	var total int64
	countQuery := h.db.Model(&models.Task{})
	if topic != "" {
		countQuery = countQuery.Where("topic_name = ?", topic)
	}
	if mqType != "" {
		countQuery = countQuery.Where("mq_type = ?", mqType)
	}
	countResult := countQuery.Count(&total)
	if countResult.Error != nil {
		response.Fail(c, http.StatusInternalServerError, "查询总数失败: "+countResult.Error.Error())
		return
	}

	response.Success(c, gin.H{
		"items":    tasks,
		"total":    total,
		"page":     page,
		"pageSize": pageSize,
	})
}

// ConsumeMessages 消费指定数量的消息
func (h *MessageHandler) ConsumeMessages(c *gin.Context) {
	var req MessageConsumeRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		response.Fail(c, http.StatusBadRequest, "无效的请求参数: "+err.Error())
		return
	}

	// 验证消费数量
	if req.Count <= 0 {
		response.Fail(c, http.StatusBadRequest, "消费数量必须大于0")
		return
	}

	// 从全局插件映射中获取对应类型的插件
	mqPlugin, ok := h.mqPlugins[req.MQType]
	if !ok {
		response.Fail(c, http.StatusBadRequest, "不支持的消息队列类型: "+string(req.MQType))
		return
	}

	// 存储消费的消息
	consumedMessages := make([]string, 0, req.Count)
	messageChannel := make(chan string, req.Count)
	errorChannel := make(chan error, 1)
	doneChannel := make(chan bool, 1)

	// 创建一个带取消功能的上下文，用于在达到消费数量后取消订阅
	ctx, cancel := context.WithCancel(c.Request.Context())
	defer cancel() // 确保在函数返回时取消上下文

	// 订阅消息
	go func() {
		// 创建一个计数器，用于跟踪已消费的消息数量
		count := 0

		// 定义消息处理函数
		handler := func(message string) error {
			if count < req.Count {
				messageChannel <- message
				count++
				if count >= req.Count {
					doneChannel <- true
					// 达到指定数量后，取消上下文，停止订阅
					cancel()
					return nil
				}
				return nil
			}
			// 如果已经达到指定数量，返回错误以停止订阅
			return errors.New("已达到指定消费数量")
		}

		// 订阅主题
		if err := mqPlugin.Subscribe(ctx, req.Topic, handler); err != nil {
			// 忽略因取消上下文导致的错误
			if ctx.Err() == nil {
				errorChannel <- err
			}
		}
	}()

	// 等待消息或错误
	select {
	case err := <-errorChannel:
		response.Fail(c, http.StatusInternalServerError, "订阅消息失败: "+err.Error())
		return
	case <-doneChannel:
		// 收集所有消息
		close(messageChannel)
		for msg := range messageChannel {
			consumedMessages = append(consumedMessages, msg)
		}

		response.Success(c, gin.H{
			"message":  "消息消费成功",
			"topic":    req.Topic,
			"mq_type":  req.MQType,
			"count":    len(consumedMessages),
			"messages": consumedMessages,
		})
	case <-ctx.Done():
		if ctx.Err() == context.Canceled && len(consumedMessages) > 0 {
			// 如果是因为达到消费数量而取消的，返回成功
			response.Success(c, gin.H{
				"message":  "消息消费成功",
				"topic":    req.Topic,
				"mq_type":  req.MQType,
				"count":    len(consumedMessages),
				"messages": consumedMessages,
			})
		} else {
			// 如果是因为超时而取消的，返回超时错误
			response.Fail(c, http.StatusRequestTimeout, "消息消费超时")
		}
		return
	}
}

// GetPendingMessages 获取队列中未消费的消息
func (h *MessageHandler) GetPendingMessages(c *gin.Context) {
	// 获取请求参数
	topic := c.Query("topic")
	mqTypeStr := c.Query("mq_type")

	if topic == "" {
		response.Fail(c, http.StatusBadRequest, "主题名称不能为空")
		return
	}

	if mqTypeStr == "" {
		response.Fail(c, http.StatusBadRequest, "消息队列类型不能为空")
		return
	}

	// 转换消息队列类型
	mqType := mq.MQType(mqTypeStr)

	// 从全局插件映射中获取对应类型的插件
	mqPlugin, ok := h.mqPlugins[mqType]
	if !ok {
		response.Fail(c, http.StatusBadRequest, "不支持的消息队列类型: "+mqTypeStr)
		return
	}

	// 获取队列统计信息
	stats, err := mqPlugin.GetQueueStats(topic)
	if err != nil {
		// 如果是主题不存在的错误，返回空统计信息
		if strings.Contains(err.Error(), "主题") && strings.Contains(err.Error(), "不存在") {
			response.Success(c, gin.H{
				"topic":             topic,
				"mq_type":           mqType,
				"pending_messages":  0,
				"consumed_messages": 0,
				"failed_messages":   0,
				"last_consumed_at":  nil,
				"exists":            false,
			})
			return
		}
		response.Fail(c, http.StatusInternalServerError, "获取队列统计信息失败: "+err.Error())
		return
	}

	// 返回队列统计信息
	response.Success(c, gin.H{
		"topic":             topic,
		"mq_type":           mqType,
		"pending_messages":  stats.PendingMessages,
		"consumed_messages": stats.ConsumedMessages,
		"failed_messages":   stats.FailedMessages,
		"last_consumed_at":  stats.LastConsumedAt,
		"exists":            true,
	})
}

// GetMessagesByStatus 获取指定topic的消息列表，支持按状态过滤和分页
func (h *MessageHandler) GetMessagesByStatus(c *gin.Context) {
	// 获取请求参数
	topic := c.Query("topic")
	mqTypeStr := c.Query("mq_type")
	statusStr := c.Query("status") // 消息状态：pending(未消费), consumed(已消费), failed(错误)
	page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
	pageSize, _ := strconv.Atoi(c.DefaultQuery("pageSize", "10"))

	if topic == "" {
		response.Fail(c, http.StatusBadRequest, "主题名称不能为空")
		return
	}

	if mqTypeStr == "" {
		response.Fail(c, http.StatusBadRequest, "消息队列类型不能为空")
		return
	}

	// 转换消息队列类型
	mqType := mq.MQType(mqTypeStr)

	// 从全局插件映射中获取对应类型的插件
	mqPlugin, ok := h.mqPlugins[mqType]
	if !ok {
		response.Fail(c, http.StatusBadRequest, "不支持的消息队列类型: "+mqTypeStr)
		return
	}

	// 获取队列统计信息
	stats, err := mqPlugin.GetQueueStats(topic)
	if err != nil {
		// 如果是主题不存在的错误，返回空列表
		if strings.Contains(err.Error(), "主题") && strings.Contains(err.Error(), "不存在") {
			response.Success(c, gin.H{
				"topic":    topic,
				"mq_type":  mqType,
				"status":   statusStr,
				"items":    []interface{}{},
				"total":    0,
				"page":     page,
				"pageSize": pageSize,
			})
			return
		}
		response.Fail(c, http.StatusInternalServerError, "获取队列统计信息失败: "+err.Error())
		return
	}

	// 根据状态获取消息列表
	var messages []interface{}
	var total int64

	switch statusStr {
	case "pending":
		// 获取未消费的消息
		pendingMessages, err := mqPlugin.GetPendingMessages(topic, page, pageSize)
		if err != nil {
			response.Fail(c, http.StatusInternalServerError, "获取未消费消息失败: "+err.Error())
			return
		}
		messages = make([]interface{}, len(pendingMessages))
		for i, msg := range pendingMessages {
			messages[i] = map[string]interface{}{
				"message":   msg,
				"status":    "pending",
				"timestamp": time.Now().Unix(), // 由于未消费消息没有时间戳，使用当前时间
			}
		}
		total = stats.PendingMessages
	case "consumed":
		// 获取已消费的消息
		consumedMessages, err := mqPlugin.GetConsumedMessages(topic, page, pageSize)
		if err != nil {
			response.Fail(c, http.StatusInternalServerError, "获取已消费消息失败: "+err.Error())
			return
		}
		messages = make([]interface{}, len(consumedMessages))
		for i, msg := range consumedMessages {
			messages[i] = map[string]interface{}{
				"message":   msg.Message,
				"status":    "consumed",
				"timestamp": msg.Timestamp,
			}
		}
		total = stats.ConsumedMessages
	case "failed":
		// 获取错误消息
		failedMessages, err := mqPlugin.GetFailedMessages(topic, page, pageSize)
		if err != nil {
			response.Fail(c, http.StatusInternalServerError, "获取错误消息失败: "+err.Error())
			return
		}
		messages = make([]interface{}, len(failedMessages))
		for i, msg := range failedMessages {
			messages[i] = map[string]interface{}{
				"message":   msg.Message,
				"status":    "failed",
				"timestamp": msg.Timestamp,
				"error":     msg.Error,
			}
		}
		total = stats.FailedMessages
	default:
		// 如果没有指定状态或状态无效，则获取所有消息
		// 这里需要分别获取三种状态的消息，然后合并
		pendingMessages, _ := mqPlugin.GetPendingMessages(topic, 1, pageSize/3)
		consumedMessages, _ := mqPlugin.GetConsumedMessages(topic, 1, pageSize/3)
		failedMessages, _ := mqPlugin.GetFailedMessages(topic, 1, pageSize/3)

		// 计算总数
		total = stats.PendingMessages + stats.ConsumedMessages + stats.FailedMessages

		// 合并消息
		messages = make([]interface{}, 0, len(pendingMessages)+len(consumedMessages)+len(failedMessages))

		// 添加未消费消息
		for _, msg := range pendingMessages {
			messages = append(messages, map[string]interface{}{
				"message":   msg,
				"status":    "pending",
				"timestamp": time.Now().Unix(),
			})
		}

		// 添加已消费消息
		for _, msg := range consumedMessages {
			messages = append(messages, map[string]interface{}{
				"message":   msg.Message,
				"status":    "consumed",
				"timestamp": msg.Timestamp,
			})
		}

		// 添加错误消息
		for _, msg := range failedMessages {
			messages = append(messages, map[string]interface{}{
				"message":   msg.Message,
				"status":    "failed",
				"timestamp": msg.Timestamp,
				"error":     msg.Error,
			})
		}
	}

	// 返回消息列表和分页信息
	response.Success(c, gin.H{
		"topic":    topic,
		"mq_type":  mqType,
		"status":   statusStr,
		"items":    messages,
		"total":    total,
		"page":     page,
		"pageSize": pageSize,
	})
}
func handleExtraConfig(extraConfigJson []byte, config map[string]interface{}) {
	var extraConfig map[string]interface{}
	if err := json.Unmarshal(extraConfigJson, &extraConfig); err == nil {
		for key, value := range extraConfig {
			config[key] = value
		}
	} else {
		log.Printf("解析额外配置失败: %v", err)
	}
}

// ListTopics 获取所有主题列表
func (h *MessageHandler) ListTopics(c *gin.Context) {
	// 获取消息队列类型参数
	mqTypeStr := c.Query("mq_type")
	if mqTypeStr == "" {
		response.Fail(c, http.StatusBadRequest, "必须指定 mq_type 参数")
		return
	}

	// 解析消息队列类型
	mqType, err := strconv.Atoi(mqTypeStr)
	if err != nil {
		response.Fail(c, http.StatusBadRequest, "无效的消息队列类型")
		return
	}

	// 从映射中获取对应类型的插件
	plugin, exists := h.mqPlugins[mq.MQType(mqType)]
	if !exists {
		response.Fail(c, http.StatusNotFound, "指定类型的消息队列插件未初始化")
		return
	}

	// 获取所有主题列表
	topics := plugin.ListTopics()

	// 返回结果
	response.Success(c, gin.H{
		"topics": topics,
		"count":  len(topics),
		"type":   plugin.Type(),
	})
}

// DeleteTopic 删除指定主题的队列
func (h *MessageHandler) DeleteTopic(c *gin.Context) {
	// 获取主题名称
	topic := c.Param("topic")
	if topic == "" {
		response.Fail(c, http.StatusBadRequest, "缺少必要的参数：topic")
		return
	}

	// 获取消息队列类型参数
	mqTypeStr := c.Query("mq_type")
	if mqTypeStr == "" {
		response.Fail(c, http.StatusBadRequest, "必须指定 mq_type 参数")
		return
	}

	// 解析消息队列类型
	mqType, err := strconv.Atoi(mqTypeStr)
	if err != nil {
		response.Fail(c, http.StatusBadRequest, "无效的消息队列类型")
		return
	}

	// 从映射中获取对应类型的插件
	plugin, exists := h.mqPlugins[mq.MQType(mqType)]
	if !exists {
		response.Fail(c, http.StatusNotFound, "指定类型的消息队列插件未初始化")
		return
	}

	// 关闭指定主题的队列
	err = plugin.CloseQueue(topic)
	if err != nil {
		response.Fail(c, http.StatusInternalServerError, "删除主题失败: "+err.Error())
		return
	}

	// 返回结果
	response.Success(c, gin.H{
		"message": "删除主题成功",
		"topic":   topic,
		"type":    plugin.Type(),
	})
}
